#ifndef _M_SERVER_H
#define _M_SERVER_H

#include <iostream>
#include <vector>
#include <string>
#include <cstring>
#include <cstdio>
#include <ctime>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <functional>
#include <sys/epoll.h>
#include <unordered_map>
#include <utility>
#include <pthread.h>
#include <mutex>
#include <thread>
#include <sys/eventfd.h>
#include <memory>
#include <sys/timerfd.h>
#include <typeinfo>
#include <cassert>
#include <thread>
#include <condition_variable>
#include <signal.h>


// 日志宏模块
enum
{
    INFO = 0,
    DEBUG,
    ERROR,
};

#define LOG_LEVEL ERROR
/* ## 需要消除可变参数为空时 的 逗号的 语法错误 */
#define LOG(level, format, ...) do{\
    if(level < LOG_LEVEL) break;\
    time_t t = time(NULL);\
    struct tm* ptm = localtime(&t);\
    char ft[16] = {'\0'};\
    strftime(ft, 15, "%H:%M:%S", ptm);\
    fprintf(stdout, "[%p %s %s:%d]" format "\n",(void*)pthread_self() , ft, __FILE__, __LINE__, ##__VA_ARGS__);\
}while(0)

#define INFO_LOG(format, ...) LOG(INFO, format, ##__VA_ARGS__)
#define DEBUG_LOG(format, ...) LOG(DEBUG, format, ##__VA_ARGS__)
#define ERROR_LOG(format, ...) LOG(ERROR, format, ##__VA_ARGS__)




// 用户态缓冲区 Buffer 模块
#define BUFFER_DEFAULT_SIZE 1024
class Buffer
{
public:
    Buffer()
        :_buffer(BUFFER_DEFAULT_SIZE), _read_idx(0), _write_idx(0)
    {}
    
    char *WritePos()        // 获取当前写入起始地址
    {
        return &(_buffer[_write_idx]);
    }

    char *ReadPos()         // 获取当前读取起始地址
    {
        return &(_buffer[_read_idx]);
    }
    const char *ReadPos() const
    {
        return &(_buffer[_read_idx]);
    }

    uint64_t HeadIdleSize()     // 获取缓冲区前面空闲空间大小
    {
        return _read_idx;
    }

    uint64_t TailIdleSize()     // 获取缓冲区后面空闲空间大小
    {
        return _buffer.size() - _write_idx;
    }

    uint64_t ReadableSize() const       // 获取缓冲区可读数据大小
    {
        return _write_idx - _read_idx;
    }

    void ReadMove(uint64_t len)     // 读偏移向后移动固定大小
    {
        if(len > ReadableSize())
        {
            // std::cerr << "读偏移移动长度，不能超过可读数据长度" << std::endl;
            LOG(ERROR, "读偏移移动长度，不能超过可读数据长度");
            return;
        }
        _read_idx += len;
    }

    void WriteMove(uint64_t len)        // 写偏移向后移动固定大小
    {
        // DEBUG_LOG("Buffer::void WriteMove(uint64_t len), len: %lld, TailIdleSize: %lld", len, TailIdleSize());
        if(len > TailIdleSize())
        {
            // std::cerr << "写偏移移动长度，不能超过 _write_idx 后的可写空间长度" << std::endl;
            LOG(ERROR, "写偏移移动长度，不能超过 _write_idx 后的可写空间长度");
            return;
        }
        _write_idx += len;
    }

    void EnsureWriteSpace(uint64_t len)     // 确保可写空间足够、不够则扩容
    {
        if(len <= TailIdleSize())       // _write_idx 后的可写空间长度足够
        {
            return;
        }
        else if(len <= TailIdleSize() + HeadIdleSize())     // 总空间足够，需要将所有可读数据移动或拷贝、到初始位置
        {
            uint64_t readablesize = ReadableSize();
            std::copy(ReadPos(), WritePos(), _buffer.begin());
            _read_idx = 0;
            _write_idx = readablesize;
        }
        else        // 总空间不够，原空间不动，直接扩容
        {
            // std::cout << "扩容前 _buffer.size() = " << _buffer.size() << std::endl;
            _buffer.resize(_write_idx + len);
            // _buffer.resize(_buffer.size() * 2);
            // std::cout << "扩容后 _buffer.size() = " << _buffer.size() << std::endl;
        }
    }


    void Write(const void *data, uint64_t len)      // 写入固定长度数据
    {
        if(len == 0) return;
        EnsureWriteSpace(len);
        std::copy((const char *)data, (const char*)data + len, WritePos());
        WriteMove(len);
    }
    void WriteString(const std::string &data)
    {
        Write((const void*)data.c_str(), data.size());
    }
    void WriteBuffer(const Buffer &data)
    {
        Write((void*)data.ReadPos(), data.ReadableSize());
    }


    void Read(void *buff, uint64_t len)     // 读取固定长度数据
    {
        if(len > ReadableSize())
            len = ReadableSize();
        std::copy(ReadPos(), ReadPos() + len, (char*)buff);
        ReadMove(len);
    }
    std::string ReadAsString(uint64_t len)
    {
        if(len > ReadableSize())
            len = ReadableSize();
        std::string str;
        str.resize(len);
        Read((void*)(&(str[0])), len);      // c_str 返回的指针是 const char*，这里不能给 const
        return str;
    }
    char *FindCRLF()        // 寻找 换行字符，为 http 准备
    {
        void *res = memchr(ReadPos(), '\n', ReadableSize());
        return (char*)res;
    }
    std::string GetLine()       // get 一个 http 请求行，包括了最后的 \n
    {
        char *pos = FindCRLF();
        if(pos == NULL)
        {
            return "";
        }
        return ReadAsString(pos - ReadPos() + 1);
    }


    void Clear()        // 清空缓冲区
    {
        _write_idx = _read_idx = 0;
    }

private:
    std::vector<char> _buffer;      // 缓冲区容器
    uint64_t _read_idx;             // 数据读取位置，64 位足够大了
    uint64_t _write_idx;            // 数据写入位置
};





// 对套接字进行封装
#define MAX_LISTEN 1024     // 最大并发连接数，用于 监听套接字的 backlog 设置

class Socket
{
private:
    int _sockfd;
public:
    int Fd()
    {
        return _sockfd;
    }
    Socket()
        :_sockfd(-1)
    {}
    Socket(int fd)
        :_sockfd(fd)
    {}
    ~Socket()
    {
        Close();
    }

    bool Create()       // 创建套接字
    {
        // int socket(int domain, int type, int protocol)
        _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if(_sockfd < 0)
        {
            LOG(ERROR, "Create Failed!");
            return false;
        }
        return true;
    }

    bool Bind(const std::string &ip, uint16_t port)     // 绑定地址信息
    {
        // int bind(int sockfd, const struct sockaddr* myaddr, socklen_t addrlen)
        struct sockaddr_in addr;        // 组织本地的 地址结构
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(addr);
        
        int ret = bind(_sockfd, (struct sockaddr*)&addr, len);
        if(ret < 0)
        {
            LOG(ERROR, "Bind Failed!");
            return false;
        }
        return true;
    }

    bool Listen(int backlog = MAX_LISTEN)       // 服务器套接字开始监听
    {
        // int listen(int sockfd, int backlog)
        int ret = listen(_sockfd, backlog);
        if(ret < 0)
        {
            LOG(ERROR, "listen Failed!");
            return false;
        }
        return true;
    }

    bool Connect(const std::string &ip, uint16_t port)      // 客户端向服务器发起连接
    {
        // int connect(int sockfd, struct sockaddr *serv_addr, int addrlen)
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());

        int ret = connect(_sockfd, (struct sockaddr*)&addr, sizeof(addr));
        if(ret != 0)    // connect 连接成功返回 0
        {
            LOG(ERROR, "connect Failed!");
            return false;
        }
        return true;
    }

    int Accept()        // 服务器获取新连接
    {
        // int accept (int sockfd, struct sockaddr *addr, socklen_t *addrlen)
        struct sockaddr_in client_addr;    // 输出型参数
        socklen_t len = sizeof(client_addr);

        int newfd = accept(_sockfd, (struct sockaddr*)&client_addr, &len);    // 此时 _sockfd 是监听套接字
        if(newfd < 0)
        {
            LOG(ERROR, "accept Failed!");
            return -1;
        }

        std::string clientip = inet_ntoa(client_addr.sin_addr);     // 解析客户端地址信息
        uint16_t clientport = ntohs(client_addr.sin_port);

        LOG(INFO, "New client connected: %s:%d , newfd: %d", clientip.c_str(), clientport, newfd);
        return newfd;
    }

    ssize_t Recv(void *buf, size_t len, int flag = 0)       // 阻塞接收, 从套接字中接收数据
    {
        // int recv(int sockfd, void *buf, int len, unsigned int flags)
        int ret = recv(_sockfd, buf, len, flag);
        if(ret <= 0)        // =0的情况，无数据可读（非阻塞情况下），或 被断开连接
        {
            // EAGAIN：接收缓冲区无数据了，非阻塞情况下返回0 并设置 errno 为EAGAIN
            // EINTER：连接被断开，被信号打断了
            if(errno == EAGAIN || errno == EINTR)
            {
                LOG(INFO, "recv nodata or signal!");
                return 0;
            }
            LOG(ERROR, "recv Failed!");
            return -1;
        }
        return ret;
    }
    ssize_t RecvNoBlock(void *buf, size_t len)  // 非阻塞接收
    {
        return Recv(buf, len, MSG_DONTWAIT);
    }

    ssize_t Send(void *buf, size_t len, int flag = 0)       // 向套接字中发送数据
    {
        if(len == 0) return 0;
        // int send(int sockfd, const void *msg, int len, int flags)
        int ret = send(_sockfd, buf, len, flag);
        if(ret < 0)
        {
            // EAGAIN：对方接收缓冲区无空间了，非阻塞情况下返回0 并设置 errno 为EAGAIN
            // EINTER：连接被断开，被信号打断了
            if(errno == EAGAIN || errno == EINTR)
            {
                LOG(INFO, "send nospace or signal!");
                return 0;
            }
            LOG(ERROR, "send Failed!");
            return -1;
        }
        return ret;
    }
    ssize_t SendNoBlock(void *buf, size_t len)  // 非阻塞发送
    {
        return Send(buf, len, MSG_DONTWAIT);
    }

    void Close()        // 关闭套接字
    {
        if(_sockfd != -1)
        {
            close(_sockfd);
            _sockfd = -1;
        }
    }

    bool CreateServer(uint16_t port, const std::string &ip = "0.0.0.0", bool block_flag = false)
    {
        // （1、创建套接字，2、绑定地址，3、开始监听，4、设置非阻塞，5、启动地址重用。。。。。获取新连接，进行通信）
        if(Create() == false) return false;
        if(block_flag) SetNonBlock();
        ReuseAddr();
        if(Bind(ip, port) == false) return false;
        if(Listen() == false) return false;
        return true;
    }

    bool CreateClient(uint16_t port, const std::string &ip)
    {
        // 创建套接字，向服务器请求连接
        if(Create() == false) return false;
        if(Connect(ip, port) == false) return false;
        return true;
    }

    void ReuseAddr()        // 开启地址端口重用
    {
        // int setsockopt(int sockfd, int level, int optname, const void *optval, socklen_t optlen);
        int val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void*)&val, sizeof(int));
        val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void*)&val, sizeof(int));
    }

    void SetNonBlock()      // 设置套接字 fd 为非阻塞
    {
        // int fcntl(int fildes, int cmd, ...);
        int flag = fcntl(_sockfd, F_GETFL, 0);  // 先获取现有属性
        fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK); // 再用 | 设置，防止直接覆盖掉原有属性
    }
};

class Poller;
class EventLoop;
class Channel
{
private:
    int _fd;
    uint32_t _events;       // 该 channel 准备需要监控的事件
    uint32_t _revents;      // 该 channel 当前已经触发的事件
    // using EventCalback = std::function<void(Channel*)>;      // 也就是说 Channel 内部的回调，由 Connection 来设置
    using EventCalback = std::function<void()>;      // 也就是说 Channel 内部的回调，由 Connection 来设置
    EventCalback _read_callback;						//可读事件触发的回调
	EventCalback _write_callback;						//可写事件触发的回调
	EventCalback _error_callback;						//错误事件触发的回调
	EventCalback _close_callback;						//连接断开事件触发的回调
	EventCalback _event_callback;						//任意事件触发的回调

    EventLoop*_loop;    // 使用 Poller 的接口操作
    // Poller *_poller;    // 使用 Poller 的接口操作

public:
    Channel(EventLoop *loop, int fd)
        :_fd(fd), _events(0), _revents(0), _loop(loop)
    {}

    int Fd()
    {
        return _fd;
    }

    uint32_t Events()    // 让 Poller 可以获取到 需要被监控什么事件
    {
        return _events;
    }

    void SetREvents(uint32_t revents)						//如果有事件就绪，EventLoop 会通过这个接口，把实际触发的事件设置进这个 Channel 的成员
    {
        _revents = revents;
    }

	void SetReadCallback(const EventCalback &cb)			//设置回调接口，用来给 外部的 Connection/EventLoop 调用的
    {
        _read_callback = cb;
    }
	void SetWriteCallback(const EventCalback &cb)
    {
        _write_callback = cb;
    }
	void SetErrorCallback(const EventCalback &cb)
    {
        _error_callback = cb;
    }
	void SetCloseCallback(const EventCalback &cb)
    {
        _close_callback = cb;
    }
	void SetEventCallback(const EventCalback &cb)
    {
        _event_callback = cb;
    }

	bool ReadAble()									    //当前是否监控了可读事件？
    {
        return (_events & EPOLLIN);
    }
	bool WriteAble()									//当前是否监控了可写事件？
    {
        return (_events & EPOLLOUT);
    }

	void EnableRead()									//启动读事件监控（将需要监控的事件添加进 成员后，调用 EventLoop 子模块 Poller 的接口进行添加事件监控）
    {
        _events |= EPOLLIN;
        // ...
        Update();
    }
	void EnableWrite()									//启动写事件监控（将需要监控的事件添加进 成员后，调用 EventLoop 子模块 Poller 的接口进行添加事件监控）
    {
        _events |= EPOLLOUT;
        // ...
        Update();
    }

	void DisableRead()									//关闭读事件监控（就是将成员 _events 的可读标志位置0 ，调用 EventLoop 子模块 Poller 的接口进行修改事件监控）
    {
        _events &= ~EPOLLIN;
        // ...
        Update();
    }
	void DisableWrite()									//关闭写事件监控（就是将成员 _events 的可写标志位置0 ，调用 EventLoop 子模块 Poller 的接口进行修改事件监控）
    {
        _events &= ~EPOLLOUT;
        // ...
        Update();
    }
	void DisableAll()									//关闭所有事件监控（就是将成员 _events 置0 ，调用 EventLoop 子模块 Poller 的接口进行修改事件监控）、但epoll 对象中的红黑树节点仍然存在
    {
        _events = 0;
        // ...
        Update();
    }

	void Remove();										//直接移除该监控，就是从 epoll 模型对象的红黑树节点中去掉
    // {
    //     // ...
    //     return _poller->RemoveEvent(this);
    // }
    // 这两个接口的实现，要放到 Poller 类de下面，否则会报错
    void Update();   // 次级封装，便于上述事件监控使用
    // {
    //     return _poller->UpdateEvent(this);
    // }

	void HandleEvent()									//  5 种事件处理，一旦链接触发事件，外部直接调用 Channel 内部的这个接口进行处理
    {
        if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI))
        {
            if(_event_callback) _event_callback();  // 读完后刷新一下链接活跃度，有点问题，读写操作都可能导致 channel 的释放
            if(_read_callback) _read_callback();
        }

        if((_revents & EPOLLOUT))
        {
            if(_event_callback) _event_callback();  // 写完后刷新一下链接活跃度，有点问题，读写操作都可能导致 channel 的释放
            if(_write_callback) _write_callback();
        }
        else if((_revents & EPOLLERR))
        {
            if(_event_callback) _event_callback();  // 先执行任意回调，再执行错误回调，否则可能出错
            if(_error_callback) _error_callback();
        }
        else if((_revents & EPOLLHUP))
        {
            if(_event_callback) _event_callback();  // 先执行任意回调，再执行错误回调，否则可能出错
            if(_close_callback) _close_callback();
        }

    }
};


#define MAX_EPOLLEVENTS 1024
class Poller
{
private:
	int _epfd;											//fd，epoll 的操作句柄
	struct epoll_event _evs[MAX_EPOLLEVENTS];			//epoll_wait的输出型参数，结构体数组，用以监控时，给 epoll wait 用以保存所有的活跃事件，然后逐个处理事件
	std::unordered_map<int, Channel*> _channels;		//通过 hash表，管理 fd 与 fd对应的监控事件管理对象 Channel 对象

private:
	void Update(Channel *channel, int op)				//次一级的封装，对 epoll 的直接操作
    {
        // int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
        int fd = channel->Fd();
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = channel->Events();      // 此 fd 需要被监控的 事件类型（读写...）
        int ret = epoll_ctl(_epfd, op, fd, &ev);
        if(ret < 0)
        {
            ERROR_LOG("Update(epoll_ctl) Failed!");
        }
        return;
    }

	bool HasChannel(Channel *channel)					//判断一个 Channel 是否已经添加了事件监控
    {
        auto it = _channels.find(channel->Fd());
        if(it == _channels.end())
        {
            return false;
        }
        return true;
    }

public:
	Poller()
        :_epfd(epoll_create(1))     // 创建 epoll 对象
    {}

	void UpdateEvent(Channel *channel)					//添加or修改 监控事件
    {
        bool ret = HasChannel(channel);
        if(ret == false)    // _channels 哈希表中不存在该 channel，进行添加
        {
            _channels.insert(std::make_pair(channel->Fd(), channel));
            return Update(channel, EPOLL_CTL_ADD);  // 向 epoll 实例中添加一个新的文件描述符
        }
        return Update(channel, EPOLL_CTL_MOD);  // 修改已存在文件描述符的事件类型，此事件类型 由 channel 的 成员 events 提供
    }

	void RemoveEvent(Channel *channel)					//移除监控：移除节点？对
    {
        auto it = _channels.find(channel->Fd());
        if(it != _channels.end())
        {
            _channels.erase(it);    // 同时从管理结构 ——  _channels 哈希表中删掉
        }
        Update(channel, EPOLL_CTL_DEL);     // 从 epoll 实例中删除一个文件描述符
    }

	void Poll(std::vector<Channel*> *activeChannels)			//开始监控，用输出型参数，返回活跃链接
    {
        // int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
        int nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1);   // _evs 是输出型参数，epoll wait 会把就绪事件，放进这个结构体数组中 每个结构体的 事件中，struct epoll_event 包含 event和data 成员
        if(nfds < 0)
        {
            if(errno == EINTR)
            {
                return; // 被信号打断
            }
            ERROR_LOG("epoll_wait error: %s\n", strerror(errno));
            abort();    // 直接退出
        }

        for(int i = 0; i < nfds; i++)   // 把epoll wait 检测到的所有就绪事件（存进_evs了） 的 链接（channel），逐个添加进输出参数中
        {
            auto it = _channels.find(_evs[i].data.fd);
            if(it == _channels.end())   // 这个就绪的channel，肯定是在 _channels哈希表中的；不然就证明之前对channel连接的管理出问题了
            {
                ERROR_LOG("Poll error!");
                abort();
            }
            it->second->SetREvents(_evs[i].events);     // 设置 epoll wait 监测到的、此 channel 实际就绪的事件，
            activeChannels->push_back(it->second);
        }
        return;
    }
};



// 定时任务的封装
// 功能：用以定时 清除 非活跃连接
using TaskFunc = std::function<void()>;

class TimerTask
{
    using ReleaseFunc = std::function<void()>;
public:    
    TimerTask(uint64_t id, uint32_t timeout, const TaskFunc &cb)        // 构造
        :_id(id), _timeout(timeout), _task_cb(cb), _canceled(false)
    {}

    ~TimerTask()        // 析构，需要间接回调任务，同时释放掉 TimerWheel 中哈希表里存储的任务对象信息
    {
        if(_canceled == false)
            _task_cb();
        _release();
    }

    void Cancel()       // 设置本定时任务对象 是否被取消
    {
        _canceled = true;
    }

    void SetRelease(const ReleaseFunc &cb)      // 设置回调
    {
        _release = cb;
    }

    uint32_t GetDelayTime()
    {
        return _timeout;
    }

private:
    uint64_t _id;		    // 定时器对象的任务id，应由外部统一为多线程分配
    uint32_t _timeout;	    // 定时任务的超时时间
    TaskFunc _task_cb;		// 需要被回调的任务，由外部传入 —— 本组件中，应当是 非活跃连接的销毁 操作
    ReleaseFunc _release;	// 释放哈希表中任务对象信息的回调
    bool _canceled;         // 该定时任务是否被取消
};



// 时间轮
// 功能：用以定时 清除 非活跃连接
class TimerWheel
{
    using PtrTask = std::shared_ptr<TimerTask>;
    using WeakTask = std::weak_ptr<TimerTask>;

private:
    int _tick;                                          // 秒针
    int _capacity;                                      // 表盘的长度，即最长的可延迟时间
    std::vector<std::vector<PtrTask>> _wheel;           // 每级时间轮都是二维数组，存储管理任务对象的智能指针
    std::unordered_map<uint64_t, WeakTask> _timers;     // 存储每个任务对象的 shared_ptr，以便刷新任务时进行拷贝；但，当该任务对象真正被析构 以回调任务时，本哈希表中对应 id 的 weak ptr 也应被删除掉；所以这个操作放在 TimerTask 的析构中，合理
    
    EventLoop *_loop;
    int _timerfd;   // 定时器的fd，库里提供的；只要 这个fd 触发一次可读，时间轮就要 tick 滴答一下；所以要对 timerfd 设置 1s 触发一次
                    // 要想监控这个 fd，得添加到 Eventloop 中，进行事件监控，并设置事件回调 —— 要读取 timerfd，并执行runtimertask
    std::unique_ptr<Channel> _timer_channel;

    static int CreateTimerfd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if(timerfd < 0)
        {
            ERROR_LOG("CreateTimerfd Failed !");
            abort();
        }

        // int timerfd_settime(int fd, int flags, const struct itimerspec *new_value, struct itimerspec *old_value);
        struct itimerspec itime;
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0;
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0;
        timerfd_settime(timerfd, 0, &itime, NULL);      // 启动定时器
        return timerfd;
    }

    int ReadTimerfd()
    {
        uint64_t times;
        int ret = read(_timerfd, &times, sizeof(times));
        if(ret < 0)
        {
            ERROR_LOG("ReadTimerfd Failed !");
            abort();
        }
        return times;
    }

    void RunTimerTask()     // 每秒钟执行一次本函数，让 tick 向后走一步，每走一步都执行对应任务
    {
        _tick = (++_tick) % _capacity;
        _wheel[_tick].clear();      // clear 可以销毁该 vector 中的所有元素，这样就达到了销毁 TimerTask 任务对象的 shared ptr 的目的，当引用计数为0时，就会回调 两个函数
    }

    void OnTime()
    {
        int times = ReadTimerfd();      // 根据实际超时次数，执行对应的超时任务
        for (size_t i = 0; i < times; i++)
        {
            RunTimerTask();
        }
        
    }

    void TimerAddInLoop(uint64_t id, uint32_t delay, const TaskFunc &cb)      // 往时间轮中添加定时任务，及其任务参数
    {
        PtrTask pt(new TimerTask(id, delay, cb));       // 实例化一个任务对象
        pt->SetRelease(std::bind(&TimerWheel::RemoveTimer, this, id));       // 设置任务对象被销毁时的回调，并绑定 this
        int pos = (_tick + delay) % _capacity;
        _wheel[pos].push_back(pt);      // 把任务对象添加到 二维时间轮中
        _timers[id] = WeakTask(pt);     // 添加到 哈希表中保存
    }

    void TimerRefreshInLoop(uint64_t id)      // 刷新或延迟定时任务，涉及到需要拷贝 原管理此对象智能指针，那么需要引入weak_ptr 对 shared_ptr 进行存储，这样并不会增加 shared_ptr 的引用计数；并引入 unordered_map 记录任务id 与 对应 weak ptr，以便需要延迟任务时，找到并进行拷贝构造
    {
        auto it = _timers.find(id);     // 从哈希表中找到保存好的 weak ptr，去构造新的 shared ptr，再根据超时时间，添加到时间轮中
        if(it != _timers.end())
        {
            // PtrTask new_pt(_timers[id]);
            PtrTask new_pt = it->second.lock();     // lock 会返回 weak ptr 管理的对象的 shared ptr
            int pos = (_tick + new_pt->GetDelayTime()) % _capacity;     // 任务被刷新后 该存放的位置
            _wheel[pos].push_back(new_pt);      // 刷新的定时任务，放入时间轮中新的位置
        }
    }

    void TimerCancelInLoop(uint64_t id)       // TimerTask 对象只有一个，改哈希表里的和 时间轮中的，一样
    {
        // DEBUG_LOG("TimerCancelInLoop begin: id: %lld", id);
        auto it = _timers.find(id);
        if(it != _timers.end())
        {
            PtrTask new_pt = it->second.lock(); // 从弱引用中还原shared_ptr，但可能失败
            if(new_pt)
            {
                new_pt->Cancel();       // 必须还原出 shared ptr 再调用 TimerTask 的方法
                // DEBUG_LOG("new_pt->Cancel() end !");
            }
            else
            {
                // DEBUG_LOG("TimerCancelInLoop: TimerTask for id %lld 已经释放", id);
                _timers.erase(it);  //此时从_timers中移除已经释放了的弱引用
            }
        }
        
    }

public:
    TimerWheel(EventLoop *loop)        // 构造，对秒针、表盘长度进行初始化
        :_tick(0), _capacity(60), _wheel(_capacity), _loop(loop), _timerfd(CreateTimerfd()), _timer_channel(new Channel(_loop, _timerfd))
    {
        _timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime, this));
        _timer_channel->EnableRead();   // 启动可读监控
    }

    // 因为定时器中有个 _timers 成员，定时器信息的操作有可能在多线程中进行，因此需要考虑线程安全问题
    // 如果不加锁（考虑效率），就需要把 对定时器的所有操作，放到同一个线程中进行，如下
    void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb);     //添加定时任务
    // {
    //     _loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop, this, id, delay, cb));
    // }

    void TimerRefresh(uint64_t id);     //刷新 or 延迟 定时任务
    // {
    //     _loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop, this, id));
    // }
    
    void TimerCancel(uint64_t id);     //取消定时任务
    // {
    //     _loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop, this, id));
    // }

    // 这个接口有线程安全问题，不能被外界用户调用 ！！！！！！
    // 只能在模块内部，在对应 EventLoop 中执行，
    bool HasTimer(uint64_t id)  // 是否有某个定时任务？
    {
        auto it = _timers.find(id);
        if(it == _timers.end())
        {
            return false;
        }
        return true;
    }

private:
    void RemoveTimer(uint64_t id)       // 为 TimerAdd 时设置回调，封装的定时任务析构需要执行的从 哈希表中移除该任务的操作
    {
        auto it = _timers.find(id);
        if(it != _timers.end())
        {
            _timers.erase(it);
        }
    }
};





// One thread One EventLoop
using Functor = std::function<void()>;					
class EventLoop
{
private:
    std::thread::id _thread_id;							//当前 EventLoop 的线程ID
    int _event_fd;										//eventfd 用以唤醒 IO 事件监控中，可能导致的阻塞，通知线程需要处理任务池中的任务了
    Poller _poller;										//通过 Poller 进行对所有描述符的事件监控
    std::unique_ptr<Channel> _event_channel;            // 通过channel 管理 eventfd 的事件，构造时 new 的 channel 交给智能指针
    
    std::vector<Functor> _tasks;						//任务池，
    std::mutex _mutex;									//用互斥锁保证 任务池操作的 线程安全

    TimerWheel _timer_wheel;                            // 定时器模块

public:
    void RunAllTask()									//执行任务池中的任务
    {
        std::vector<Functor> functor;       // 只需要对本线程取出任务时进行加锁，提高效率
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.swap(functor);
        }
        for(auto &f : functor)
        {
            f();
        }
        return;
    }

    static int CreateEventfd()
    {
        // int eventfd(unsigned int initval, int flags);
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);   //在 exec 调用时自动关闭文件描述符 + 非阻塞
        if(efd < 0)
        {
            ERROR_LOG("CreateEventfd Failed !");
            abort();    // 程序异常退出
        }
        return efd;
    }

    void ReadEventfd()      // eventfd 的读事件回调
    {
        uint64_t res = 0;
        int ret = read(_event_fd, &res, sizeof(res));   // 读出来：其实就是清空 eventfd 的计数器，不读的话，eventfd 岂不是一直可读触发
        if(ret < 0)
        {
            if(errno == EINTR || errno == EAGAIN)   // 被信号打断或者 无数据可读
            {
                return;
            }
            ERROR_LOG("ReadEventfd Failed !");
            abort();
        }
        return;
    }

    void WakeupEventfd()       //通过 eventfd 唤醒 阻塞的 epoll，提醒 执行 任务池中的任务
    {
        // 其实就是 eventfd 写入一个数据，eventfd 就会触发可读事件，让 epoll 不再阻塞，接着 Poll 往下执行，然后runtask
        uint64_t val = 1;   // 8字节数据
        int ret = write(_event_fd, &val, sizeof(val));
        if(ret < 0)
        {
            if(errno == EINTR)
            {
                return;
            }
            ERROR_LOG("WakeupEventfd Failed !");
            abort();
        }
        return;
    }

public:
    EventLoop()                                         // 管理 eventfd、设置可读事件回调读取 event事件通知次数、、启动对 eventfd 的读事件监控
        :_thread_id(std::this_thread::get_id())
        ,_event_fd(CreateEventfd())
        ,_event_channel(new Channel(this, _event_fd))   // 传入 EventLoop 对象指针，通过 EventLoop 去间接调 Channel 和 Poller 中的方法
        // ,_event_channel(new Channel(&_poller, _event_fd))
        ,_timer_wheel(this)
    {
        _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventfd, this));    // 给eventfd 设置的读事件回调，用以读取 eventfd 的事件通知次数
        _event_channel->EnableRead();   // 启动对 eventfd 的读事件监控
    }

    void Start()			//三步：开始事件监控，对就绪事件进行处理，执行任务池中的任务
    {
        while(true)
        {
            std::vector<Channel *> acts;    
            _poller.Poll(&acts);        // 开始监控并获取就绪连接

            for(auto &channel : acts)       // 逐个处理 epoll wait 出来的就绪 fd，对应的 channel
            {
                channel->HandleEvent();     // 每个事件就绪的 channel 都去调用，以处理各自事件
            }

            RunAllTask();   // 执行任务池中的任务
        }
    }

    bool IsInLoop()									    //用于判断当前线程是否是EventLoop对应的线程
    {
        return (_thread_id == std::this_thread::get_id());
    }

    void AssertInLoop() // 如果不在本线程内，就不要执行
    {
        assert(_thread_id == std::this_thread::get_id());
    }

    void RunInLoop(const Functor &cb)					//判断将要执行的任务是否处于当前线程中，如果是则执行，不是就压入任务池
    {
        if(IsInLoop())
        {
            return cb();    // 将要执行的任务处于当前线程，直接执行
        }
        return PoolInLoop(cb);  // 将要执行的任务不处于当前线程，压入任务池
    }

    void PoolInLoop(const Functor &cb)					//将操作push 进 任务池
    {
        DEBUG_LOG("A Task is pushed in Pool !!! Functor: %p", cb);
        {
            std::unique_lock<std::mutex> _lock(_mutex); // 访问共享资源 _tasks
            _tasks.push_back(cb);
        }
        // 压入任务完成后，需要唤醒 可能因没有事件就绪，而导致的 epoll 阻塞，用 eventfd 唤醒？
        // 其实就是向 eventfd 写入一个数据，eventfd 就会触发可读事件，让 epoll 不再阻塞，接着 Poll 往下执行，然后runtask
        WakeupEventfd();
    }

    void UpdateEvent(Channel *channel)					//添加 or 修改描述符的 事件监控类型
    {
        _poller.UpdateEvent(channel);
    }
    void RemoveEvent(Channel *channel)					//移除描述符的监控
    {
        _poller.RemoveEvent(channel);
    }
    

    void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb)
    {
        return _timer_wheel.TimerAdd(id, delay, cb);
    }
    void TimerRefresh(uint64_t id)
    {
        return _timer_wheel.TimerRefresh(id);
    }
    void TimerCancel(uint64_t id)
    {
        return _timer_wheel.TimerCancel(id);
    }
    bool HasTimer(uint64_t id)
    {
        return _timer_wheel.HasTimer(id);
    }

};


// 关联 EventLoop 和 Thread
class LoopThread
{
private:
    // 互斥锁和条件变量：用于实现获取 _loop 的同步关系，避免线程已经创建了但是 _loop 还未实例化之前，就去获取 _loop
    std::mutex _mutex;      // 互斥锁
    std::condition_variable _cond;       // 条件变量
    EventLoop *_loop;       // 这个对象需要在线程内实例化
    std::thread _thread;    // EventLoop 对应的线程

private:
    void ThreadEntry()      // 线程的入口函数，需要实例化 EventLoop 对象，并且唤醒可能阻塞在_cond 上的线程，并运行EventLoop模块的功能—— 即 Start（）
    {
        EventLoop loop;     // 让loop的生命周期，跟随本线程；且这个loop的id就是本线程id
        {
            std::unique_lock<std::mutex> lock(_mutex);  // 加锁，保证 线程创建与获取loop的同步
            _loop = &loop;          
            _cond.notify_all();     // 唤醒阻塞在 _cond 下的线程
        }
        loop.Start();   // 循环执行，监控 fd，并处理事件
    }

public:
    LoopThread()    // 创建线程，设定线程的入口函数
        :_loop(nullptr), _thread(std::thread(&LoopThread::ThreadEntry, this))
    {}

    EventLoop *GetLoop()    // 返回与当前线程关联的 EventLoop 对象指针
    {
        EventLoop *loop = nullptr;  // 保证loop的安全操作，用个临时变量
        {
            // 保证线程的同步，避免线程已经创建了但是 _loop 还未实例化之前，就去获取 _loop
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [&](){return _loop != nullptr;});   // loop 为空，一直阻塞
            loop = _loop;
        }
        return loop;
    }
};





// 可以存任何数据的 容器类
class Any
{
public:
    Any()
        :_content(nullptr)
    {}

    ~Any()
    {
        delete _content;
    }

    template<class T>
    Any(const T &val)
        :_content(new placeholder<T>(val))
    {}

    Any(const Any &other)
        :_content(other._content ? other._content->clone() : nullptr)
    {}

    void swap(Any &other)
    {
        std::swap(_content, other._content);
    }

    template<class T>
    Any &operator=(const T &val)
    {
        Any tmp(val);
        swap(tmp);
        return *this;
    }

    Any &operator=(Any other)
    {
        swap(other);
        return *this;
    }

    template<class T>
    T *get()
    {
        if(typeid(T) != _content->type())       // 获取数据的类型必须与保存的一致
            return nullptr;
        return &(((placeholder<T> *)_content)->_val);
    }

private:
    class holder
    {
    public:
        virtual ~holder() {};
        virtual const std::type_info& type() = 0;
        virtual holder *clone() = 0;
    };

    template<class T>
    class placeholder : public holder
    {
    public:
        placeholder(const T& val)
            :_val(val)
        {}

        virtual ~placeholder()
        {}

        virtual const std::type_info& type()
        {
            return typeid(T);
        }

        virtual holder *clone()
        {
            return (new placeholder<T>(_val));
        }

        T _val;
    };

    holder *_content;
};


class Connection;
// DISCONNECTED 连接关闭状态
// CONNECTING   连接建立成功，待设置状态
// CONNECTED    连接建立完成，各种设置已完成，可以进行通信
// DISCONNECTING 待关闭状态
typedef enum
{
    DISCONNECTED,
    CONNECTING,
    CONNECTED,
    DISCONNECTING
}ConnStatu;
using PtrConnection = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>        // 连接的管理类，TcpServer 类中会对多个连接的 shared_ptr 进行管理
{
private:
    uint64_t _conn_id;      // 连接的唯一 id，便于对连接的管理
    // uint64_t _timer_id;  定时器id，也必须唯一，这里复用 连接id 作为定时器id
    int _sockfd;            // 此连接 关联的 fd
    bool _enable_inactive_release;      // 是否启动非活跃连接销毁，默认 false
    EventLoop *_loop;       // 此连接绑定的 loop，
    ConnStatu _statu;       // 此连接的状态
    Socket _socket;         // 此连接的 套接字操作管理
    Channel _channel;       // 此连接的 事件管理
    Buffer _in_buffer;      //  输入缓冲区，存放从套接字中读取的数据
    Buffer _out_buffer;     //  输出缓冲区，存放需要向 套接字中发送的数据
    Any _context;           // 对用户层协议 请求的 接收数据的 处理上下文，记录当前处理的阶段

    // 四个 连接不同阶段的 回调函数，需要被 TcpServer 类进行设置，而 TcpServer 中的设置，需要被 用户设置
    using ConnectedCallback = std::function<void(const PtrConnection&)>;
    using MessageCallback = std::function<void(const PtrConnection&, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConnection&)>;
    using AnyEventCallback = std::function<void(const PtrConnection&)>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _anyevent_callback;

    // 因为 TcpServer 类中会对多个连接的 shared_ptr 进行管理， 当某个连接关闭时，需要从 TcpServer 类的管理容器中删除
    ClosedCallback _server_closed_callback;

private:
    // 5个channel 的事件回调函数
    void HandleRead()       // 本连接 fd 触发可读事件后调用，接收 socket 数据放入缓冲区，然后调用_message_callback进行业务处理
    {
        char buf[65536];
        ssize_t ret = _socket.RecvNoBlock(buf, sizeof(buf));
        if(ret < 0)
        {
            // 读取出错，但不能立刻关闭连接，需要检查输入输出 Buffer 是否有待处理数据，进行处理后再关闭连接
            return ShutDownInLoop();
        }
        _in_buffer.Write(buf, ret);     // ret 为0就不写了

        if(_in_buffer.ReadableSize() > 0)
        {
            return _message_callback(shared_from_this(), &_in_buffer);
        }
    }

    void HandleWrite()      // 本连接 fd 触发可写事件后调用，将发送缓冲区中的数据 send ——  实际 send 数据的接口
    {
        ssize_t ret = _socket.SendNoBlock(_out_buffer.ReadPos(), _out_buffer.ReadableSize());   // out buffer中的可读数据，就是待发送数据
        if(ret < 0)
        {
            // 发送错误说明 对方的接收缓冲区满了 或者 对方关闭连接了？直接处理自己接收缓冲区的剩余数据，然后关闭连接？
            if(_in_buffer.ReadableSize() > 0)
            {
                _message_callback(shared_from_this(), &_in_buffer);
            }
            return Release();
        }

        // 发送成功
        _out_buffer.ReadMove(ret);  // buffer 的可读数据减少
        if(_out_buffer.ReadableSize() == 0)
        {
            _channel.DisableWrite();    // 发送缓冲区无数据了，关闭可写监控，避免可写事件可能一直触发，造成 loop busy
            if(_statu == DISCONNECTING)
            {
                return Release(); // 如果刚好还是待关闭，那就直接关闭
            }
        }
        return;
    }

    void HandleClose()      // 本链接 fd 触发挂断事件后调用，调用 _closed_callback
    {
        // 连接挂断，套接字什么都做不了，只能 处理数据 + 关闭连接
        if(_in_buffer.ReadableSize() > 0)
        {
            _message_callback(shared_from_this(), &_in_buffer);
        }
        // _closed_callback??
        return Release();
    }
    void HandleError()      // 本连接 fd 触发 Error事件后调用
    {
        return HandleClose();
    }

    void HandleEvent()      // 本连接 fd 触发任意事件的调用，_anyevent_callback
    {
        // 刷新连接活跃度 + _anyevent_callback
        if(_enable_inactive_release == true)
        {
            _loop->TimerRefresh(_conn_id);
        }

        if(_anyevent_callback)
        {
            _anyevent_callback(shared_from_this());
        }
    }

    void ConnectedInLoop()  // 连接建立之后，要进行各种设置：修改连接状态，为channel设置事件回调，启动事件监控
    {
        if(_statu != CONNECTING)
        {
            ERROR_LOG("ConnectedInLoop, _statu != CONNECTING");
            abort();
        }
        _statu = CONNECTED;
        // 启动读事件监控，是在构造之后进行（在构造中的 是否启动非活跃连接销毁，之后进行）
        _channel.EnableRead();
        if(_connected_callback) _connected_callback(shared_from_this());
    }

    void ReleaseInLoop()    // 实际的释放连接接口
    {
        // DEBUG_LOG("ReleaseInLoop [start] conn_id=%llu, fd=%d, status=%d", _conn_id, _sockfd, _statu);

        // 修改连接状态 + 移除连接的事件监控 + 关闭描述符 + 如果当前定时器队列中还有该连接的定时销毁任务，则取消任务 + 调用关闭回调
        _statu = DISCONNECTED;
        // DEBUG_LOG("ReleaseInLoop: set status to DISCONNECTED");

        _channel.Remove();
        // DEBUG_LOG("ReleaseInLoop: channel removed for fd=%d", _sockfd);

        _socket.Close();
        // DEBUG_LOG("ReleaseInLoop: socket closed, fd=%d", _sockfd);

        if(_loop->HasTimer(_conn_id))
        {
            // DEBUG_LOG("ReleaseInLoop: canceling timer for conn_id=%llu", _conn_id);
            CancelInactiveReleaseInLoop();
        }
        else
        {
            // DEBUG_LOG("ReleaseInLoop: no timer to cancel for conn_id=%llu", _conn_id);
        }

        if(_closed_callback)
        {
            // DEBUG_LOG("ReleaseInLoop: calling _closed_callback for conn_id=%llu", _conn_id);
            _closed_callback(shared_from_this());    // 先调用户的 close 回调，避免 TcpServer 设置的回调直接释放连接后，用户调不了了
        }
        if(_server_closed_callback)
        {
            // DEBUG_LOG("ReleaseInLoop: calling _server_closed_callback for conn_id=%llu", _conn_id);
            _server_closed_callback(shared_from_this());  // 再调 TcpServer 设置的close 回调
        }

        // DEBUG_LOG("ReleaseInLoop [end]");
    }

    void SendInLoop(Buffer &buf)     // 将数据放入发送buffer 的接口，启动写事件监控 —— 当触发可写后，会回调HandleWrite。可写回调就是channel中设置的，应该是要在构造时设置！！！
    {
        if(_statu == DISCONNECTED) return;
        _out_buffer.WriteBuffer(buf);
        if(_channel.WriteAble() == false)
        {
            _channel.EnableWrite(); // 如果没开启写事件监控，就开启
        }
    }

    void ShutDownInLoop()   // 连接关闭接口：并不是立刻关闭，先检查还有无数据待处理
    {
        _statu = DISCONNECTING; // 待关闭状态，接下来检查
        if(_in_buffer.ReadableSize() > 0)
        {
            if(_message_callback) _message_callback(shared_from_this(), &_in_buffer);
        }
        if(_out_buffer.ReadableSize() > 0)
        {
            if(_channel.WriteAble() == false)
            {
                _channel.EnableWrite();
            }
        }
        if(_out_buffer.ReadableSize() == 0)     // 是不是应该发送一次就直接 释放？这里是等发送完了再释放。。。
        {
            Release();
        }
    }

    void EnableInactiveReleaseInLoop(int sec)      // 启动非活跃连接销毁
    {
        // 1、设置标志位
        _enable_inactive_release = true;
        // 2、添加定时销毁任务：如果当前定时销毁任务已经存在，就刷新；不存在就新增
        if(_loop->HasTimer(_conn_id))
        {
            return _loop->TimerRefresh(_conn_id);
        }
        else
        {
            return _loop->TimerAdd(_conn_id, sec, std::bind(&Connection::Release, this));
        }
    }

    void CancelInactiveReleaseInLoop()      // 取消非活跃连接销毁
    {
        // DEBUG_LOG("CancelInactiveReleaseInLoop: conn_id=%llu", _conn_id);
        // 1、设置标志位
        _enable_inactive_release = false;
        // 2、存在就取消
        if(_loop->HasTimer(_conn_id))
        {
            // DEBUG_LOG("CancelInactiveReleaseInLoop: canceling timer for conn_id=%llu", _conn_id);
            return _loop->TimerCancel(_conn_id);
        }
        else
        {
            // DEBUG_LOG("CancelInactiveReleaseInLoop: no timer to cancel for conn_id=%llu", _conn_id);
        }
    }

    // 切换协议：更新连接上下文 及 回调
    void UpdateInLoop(const Any &context, const ConnectedCallback& conn, const MessageCallback& msg, const ClosedCallback &closed, const AnyEventCallback &event)
    {
        // SetContext(context);
        _context = context;
        _connected_callback = conn;
        _message_callback = msg;
        _closed_callback = closed;
        _anyevent_callback = event;
    }

public:
    Connection(EventLoop *loop, uint64_t conn_id, int sockfd)
        :_loop(loop), _conn_id(conn_id), _sockfd(sockfd), _enable_inactive_release(false), 
         _statu(CONNECTING), _socket(sockfd), _channel(loop, _sockfd)
    {
        _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
        _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
        _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
    }

    ~Connection()
    {
        DEBUG_LOG("Release Connection: %p", this);
    }

    int Fd()
    {
        return _sockfd;
    }

    int Id()        // 连接id
    {
        return _conn_id;
    }

    bool IsConnected()      // 是否处于 CONNECTED
    {
        return (_statu == CONNECTED);
    }

    void SetContext(const Any& context)       // 设置上下文，连接建立完成时调用
    {
        _context = context;
    }

    Any *GetContext()       // 获取此 Any 容器的指针，上下文内容
    {
        return &_context;
    }

    void SetConnectedCallback(const ConnectedCallback &cb)
    {
        _connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback &cb)
    {
        _message_callback = cb;
    }
    void SetClosedCallback(const ClosedCallback &cb)
    {
        _closed_callback = cb;
    }
    void SetAnyEventCallback(const AnyEventCallback &cb)
    {
        _anyevent_callback = cb;
    }
    void SetSrvClosedCallback(const ClosedCallback &cb)
    {
        _server_closed_callback = cb;
    }

    void Connected()        // 连接待设置状态：需要设置channel回调，启动事件监控，调用_connected_callback
    {
        _loop->RunInLoop(std::bind(&Connection::ConnectedInLoop, this));
    }

    void Send(const char *data, size_t len)       // 发送数据，将数据放入发送buffer，启动写事件监控，可写回调就是channel中设置的！！！
    {
        // 用户传入的data，可能是个临时空间，而这个发送任务可能不会立即执行，而是被 push 进了任务池
        // 这样用户传入的data，到此任务被执行时 可能已经释放，所以需要临时保存
        Buffer buf;
        buf.Write(data, len);
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));
    }

    void ShutDown()         // 提供给用户的连接关闭接口：并不是立刻关闭，先检查还有无数据待处理
    {
        _loop->RunInLoop(std::bind(&Connection::ShutDownInLoop, this));
    }
    void Release()  // 释放操作应该被压入任务池
    {
        _loop->PoolInLoop(std::bind(&Connection::ReleaseInLoop, this));
    }

    void EnableInactiveRelease(int sec)      // 启动非活跃连接销毁，sec 秒后销毁
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }

    void CancelInactiveRelease()           // 取消非活跃连接销毁
    {
        _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }

    // 切换协议：需要重置上下文和 阶段性回调函数 —— 非线程安全 —— 用户上层调用此接口时，必须立即执行本函数进行切换，以使用新协议处理后续任务
    // 这个接口必须在 loop 线程中立即执行，不能被压入任务队列中
    void Update(const Any &context, const ConnectedCallback& conn, const MessageCallback& msg, const ClosedCallback &closed, const AnyEventCallback &event)
    {
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::UpdateInLoop, this, context, conn, msg, closed, event));
    }
};



// 对监听套接字的管理
class Acceptor
{
private:
    Socket _socket;     // 创建监听套接字
    EventLoop *_loop;    // 对监听套接字进行事件监控管理
    Channel _channel;    // 对监听套接字进行事件管理
    
    using AcceptCallback = std::function<void(int)>;
    AcceptCallback _accept_callback;

private:
    void HandleRead()       // 监听套接字的可读回调
    {
        // DEBUG_LOG("Acceptor::HandleRead begin !!!");
        int newfd = _socket.Accept();
        if(newfd < 0)
        {
            // ERROR_LOG("Acceptor::HandleRead Failed !");
            return;
        }
        if(_accept_callback) _accept_callback(newfd);
        return;
    }

    int CreateSrv(uint16_t port)
    {
        assert(_socket.CreateServer(port));
        return _socket.Fd();
    }

public:
    Acceptor(EventLoop *loop, uint16_t port)
        :_socket(CreateSrv(port)), _loop(loop), _channel(loop, _socket.Fd())
    {
        _channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));

        // 启动可读事件监控不能放在构造中，否则容易导致 在用户未设置可读事件的上层回调之前，已经触发了可读事件
        // _channel.EnableRead();
    }

    void Listen()
    {
        _channel.EnableRead();  // 和 套接字的listen 不冲突吧？这里是启动可读监控，把 fd 挂到 epoll 下面了
    }

    void SetAcceptCallback(const AcceptCallback &cb)
    {
        _accept_callback = cb;
    }
};


// 线程池
class LoopThreadPool
{
private:
    int _thread_count;      // 从属线程的数量
    int _next_idx;     // 下一个要分配的线程索引
    EventLoop *_baseloop;   // 主 EventLoop，运行在主线程，从属线程数量如果为0，则所有操作都在 baseloop 中
    std::vector<LoopThread*> _threads;       // 保存所有的LoopThread 对象
    std::vector<EventLoop*> _loops;     // 如果从属线程数量 >0 ，则从 _loops 中进行分配

public:
    LoopThreadPool(EventLoop *baseloop)
        :_thread_count(0), _next_idx(0), _baseloop(baseloop)
    {}

    void SetThreadCount(int count)
    {
        _thread_count = count;
    }

    void Create()       // 创建从属线程
    {
        if(_thread_count > 0)
        {
            _threads.resize(_thread_count);
            _loops.resize(_thread_count);
            for(int i = 0; i < _thread_count; i++)
            {
                _threads[i] = new LoopThread();
                _loops[i] = _threads[i]->GetLoop();
            }
        }
        return;
    }

    EventLoop *NextLoop()
    {
        if(_thread_count == 0)      // 如果没有从属线程
        {
            return _baseloop;
        }
        _next_idx = (_next_idx + 1) % _thread_count;
        return _loops[_next_idx];
    }
};



class TcpServer
{
private:
    uint16_t _port;
    uint64_t _next_id;      // 增长的id
    int _timeout;           // 非活跃连接释放的时间
    bool _enable_inactive_release;      // 是否启动了非活跃连接超时销毁
    EventLoop _baseloop;    // 主线程的 EventLoop 对象，负责监听事件的管理
    Acceptor _acceptor;     // 监听套接字的管理对象
    LoopThreadPool _pool;   // 这是从属 EventLoop 线程池
    std::unordered_map<uint64_t, PtrConnection> _conns;     // 管理所有连接与其对应的 shared_ptr 对象，这里不是弱引用，注意！！！

    // 四个 连接不同阶段的 回调函数，需要被 TcpServer 类进行设置，而 TcpServer 中的设置，需要被 用户设置
    using ConnectedCallback = std::function<void(const PtrConnection&)>;
    using MessageCallback = std::function<void(const PtrConnection&, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConnection&)>;
    using AnyEventCallback = std::function<void(const PtrConnection&)>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _anyevent_callback;

private:
    void NewConnection(int fd)      // 为新连接构造一个 Connection 进行管理
    {
        DEBUG_LOG("NewConnection begin!!!");
        _next_id++;
        PtrConnection conn(new Connection(_pool.NextLoop(), _next_id, fd));
        conn->SetMessageCallback(_message_callback);  
        conn->SetConnectedCallback(_connected_callback);
        conn->SetClosedCallback(_closed_callback);
        conn->SetAnyEventCallback(_anyevent_callback);
        conn->SetSrvClosedCallback(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));

        if(_enable_inactive_release) conn->EnableInactiveRelease(_timeout);
        conn->Connected();  // 就绪初始化，挂到从属线程上处理事件

        _conns.insert(std::make_pair(_next_id, conn));
    }

    void RemoveConnectionInLoop(const PtrConnection &conn)         // 从 _conns 中移除连接信息，否则该连接不会释放，因为存的是 shared ptr 非 weak ptr
    {
        int id = conn->Id();
        auto it = _conns.find(id);
        if(it != _conns.end())
        {
            _conns.erase(it);
        }
    }

    void RemoveConnection(const PtrConnection &conn)         // 从 _conns 中移除连接信息，否则该连接不会释放，因为存的是 shared ptr 非 weak ptr
    {
        return _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, conn));
    }

    void RunAfterInLoop(const TaskFunc &task, int delay)       // 用于用户添加自定义的定时任务
    {
        _next_id++;
        _baseloop.TimerAdd(_next_id, delay, task);
    }


public:
    TcpServer(uint16_t port)
        :_port(port), _next_id(0), _enable_inactive_release(false), _acceptor(&_baseloop, port)
        ,_pool(&_baseloop)
        {
            _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
            _acceptor.Listen();     // 将监听套接字挂到 baseloop 上
        }

    void SetThreadCount(int count)
    {
        return _pool.SetThreadCount(count);
    }

    void SetConnectedCallback(const ConnectedCallback &cb)
    {
        _connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback &cb)
    {
        _message_callback = cb;
    }
    void SetClosedCallback(const ClosedCallback &cb)
    {
        _closed_callback = cb;
    }
    void SetAnyEventCallback(const AnyEventCallback &cb)
    {
        _anyevent_callback = cb;
    }

    void EnableInactiveRelease(int timeout)
    {
        _timeout = timeout;
        _enable_inactive_release = true;
    }
    
    void RunAfter(const TaskFunc &task, int delay)       // 用于用户添加自定义的定时任务
    {
        return _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, task, delay));
    }

    void Start()
    {
        _pool.Create();     // 创建线程池中的从属线程
        return _baseloop.Start();   //开始监控，并处理事件
    }
};


// 当连接断开后，继续 send 会触发异常信号  SIGPIPE，这里忽略掉，为了不让进程退出
class NetWork
{
public:
    NetWork()
    {
        DEBUG_LOG("SIGPIPE INIT");
        signal(SIGPIPE, SIG_IGN);
    }
};
static NetWork nw;



void Channel::Remove()										//直接移除该监控，就是从 epoll 模型对象的红黑树节点中去掉
{
    // ...
    return _loop->RemoveEvent(this);
}

void Channel::Update()   // 次级封装，便于上述事件监控使用
{
    return _loop->UpdateEvent(this);
}

void TimerWheel::TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop, this, id, delay, cb));
}

void TimerWheel::TimerRefresh(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop, this, id));
}

void TimerWheel::TimerCancel(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop, this, id));
}

#endif